package com.wzf.greattruth.zookeeper.operator;

import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * zookeeper的操作器
 * 
 * @author wuzhengfei
 *
 * @param <T>
 */
public abstract class ZooKeeperOperator<P, T> {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperOperator.class);

    private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
    private ZooKeeper zookeeper;
    private String dir;
    private byte[] datas = null;
    private CreateMode mode;

    private int retryTimes;
    private long retryDelay;

    /**
     * 执行一个zk操作
     * 
     * @return
     *
     * @return the result of the operation or null
     * @throws KeeperException
     * @throws InterruptedException
     */
    protected abstract T doOperate(P operateParam) throws KeeperException, InterruptedException;

    public ZooKeeperOperator(ZooKeeper zookeeper, String dir) {
        super();
        this.zookeeper = zookeeper;
        this.dir = dir;
    }

    public ZooKeeperOperator(ZooKeeper zookeeper, String dir, byte[] datas, CreateMode mode) {
        super();
        this.zookeeper = zookeeper;
        this.dir = dir;
        this.datas = datas;
        this.mode = mode;
    }

    public ZooKeeperOperator(ZooKeeper zookeeper, String dir, byte[] datas, CreateMode mode, int retryTimes,
            long retryDelay) {
        super();
        this.zookeeper = zookeeper;
        this.dir = dir;
        this.datas = datas;
        this.mode = mode;
        this.retryTimes = retryTimes;
        this.retryDelay = retryDelay;
    }

    /**
     * 执行，如果需要的话，再失败时进行重试
     * 
     * @param retryTimes
     * @param retryDelay
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    public T operate(P operateParam) throws KeeperException, InterruptedException {
        KeeperException exception = null;
        int times = getRetryTimes() + 1;
        for (int i = 0; i < times; i++) {
            try {
                return doOperate(operateParam);
            } catch (KeeperException.SessionExpiredException e) {
                LOGGER.error("Exception! currentTimes={}", i + 1, e);
                throw e;
            } catch (KeeperException.ConnectionLossException e) {
                LOGGER.info("ConnectionLossException, but it will retry in {}ms, currentTimes={}, maxRetryTimes={}",
                        retryDelay, i + 1, times, e);
                long retryDelay = getRetryDelay();
                if (retryDelay > 0) {
                    try {
                        Thread.sleep(retryDelay);
                    } catch (InterruptedException e1) {
                        LOGGER.error("InterruptedException! currentTimes={}, maxRetryTimes={}", i + 1, times, e);
                        throw e1;
                    }
                }
                exception = e;
            }
        }
        throw exception;
    }

    public int getRetryTimes() {
        return retryTimes;
    }

    public void setRetryTimes(int retryTimes) {
        this.retryTimes = retryTimes;
    }

    public long getRetryDelay() {
        return retryDelay;
    }

    public void setRetryDelay(long retryDelay) {
        this.retryDelay = retryDelay;
    }

    public byte[] getDatas() {
        return datas;
    }

    public void setDatas(byte[] datas) {
        this.datas = datas;
    }

    public CreateMode getMode() {
        return mode;
    }

    public void setMode(CreateMode mode) {
        this.mode = mode;
    }

    public ZooKeeper getZookeeper() {
        return zookeeper;
    }

    public void setZookeeper(ZooKeeper zookeeper) {
        this.zookeeper = zookeeper;
    }

    public String getDir() {
        return dir;
    }

    public void setDir(String dir) {
        this.dir = dir;
    }

    public List<ACL> getAcl() {
        return acl;
    }

    public void setAcl(List<ACL> acl) {
        this.acl = acl;
    }
}
